package cn.xiaobai.spring.boot.domain;

import lombok.Data;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.*;

/**
 * kafka 配置简化
 *
 * @Author yangdaji
 * @Date 2023/11/4
 * @PackageName cn.xiaobai.domain
 * @ClassName KafkaProperties
 */
@Data
@ConfigurationProperties(prefix = "kafka-plus")
public class KafkaPlusProperties {

    private List<String> bootstrapServers = new ArrayList(Collections.singletonList("localhost:9092"));
    private final Map<String, String> properties = new HashMap();
    private final Consumer consumer = new Consumer();
    private final Producer producer = new Producer();
    private final Listener listener = new Listener();

    public Map<String, Object> buildCommonProperties() {
        Map<String, Object> properties = new HashMap<>();
        if (this.bootstrapServers != null) {
            properties.put("bootstrap.servers", this.bootstrapServers);
        }

        if (!CollectionUtils.isEmpty(this.properties)) {
            properties.putAll(this.properties);
        }

        return properties;
    }

    public Map<String, Object> buildConsumerProperties() {
        Map<String, Object> properties = buildCommonProperties();
        Map<String, Object> consumerMap = consumer.buildProperties();
        mergeSameValeToMap(properties, consumerMap);
        return properties;
    }

    public Map<String, Object> buildProducerProperties() {
        Map<String, Object> properties = buildCommonProperties();
        Map<String, Object> producerMap = producer.buildProperties();
        mergeSameValeToMap(properties, producerMap);
        return properties;
    }

    private void mergeSameValeToMap(Map<String, Object> commonMap, Map<String, Object> map) {
        map.forEach((at, val) -> {
            if (val != null) {
                commonMap.put(at, val);
            }
        });
    }


    @Data
    public static class Consumer {
        private Integer autoCommitInterval = 5000;
        private String autoOffsetReset = RulesType.LATEST.getName();
        private List<String> bootstrapServers;
        private String clientId;
        private Boolean enableAutoCommit = true;
        private Duration fetchMaxWait = Duration.ofMillis(500);
        private Integer fetchMaxSize = 50 * 1024 * 1024;
        private Integer fetchMinSize = 10 * 1024 * 1024;
        private String groupId;
        private Class<?> keyDeserializer = StringDeserializer.class;
        private Class<?> valueDeserializer = StringDeserializer.class;
        private Integer maxPollRecords = 500;
        private Duration maxPollInterval = Duration.ofMillis(30 * 1000);
        private final Map<String, String> properties = new HashMap<>();


        public Map<String, Object> buildProperties() {
            Map<String, Object> properties = new HashMap<>();
            properties.put("bootstrap.servers", this.getBootstrapServers());
            properties.put("group.id", this.getGroupId());
            properties.put("enable.auto.commit", this.getEnableAutoCommit());
            properties.put("auto.offset.reset", this.getAutoOffsetReset());
            properties.put("max.poll.records", this.getMaxPollRecords());
            properties.put("auto.commit.interval.ms", this.getAutoCommitInterval());
            properties.put("max.poll.interval.ms", (int) this.getMaxPollInterval().toMillis());
            properties.put("fetch.max.wait.ms", (int) this.getFetchMaxWait().toMillis());
            properties.put("fetch.max.bytes", this.getFetchMaxSize());
            properties.put("fetch.min.bytes", this.getFetchMinSize());
            properties.put("key.deserializer", this.getKeyDeserializer());
            properties.put("value.deserializer", this.getValueDeserializer());
            properties.putAll(this.getProperties());
            return properties;
        }

        public enum RulesType {
            /**
             * 自动重置偏移量到最早的偏移量
             */
            EARLIEST("earliest"),
            /**
             * 默认，自动重置偏移量为最新的偏移量
             */
            LATEST("latest"),
            /**
             * 如果消费组原来的偏移量不存在，则向消费者抛异常
             */
            NONE("none");

            private String name;

            RulesType(String name) {
                this.name = name;
            }

            public String getName() {
                return name;
            }
        }
    }

    @Data
    public static class Producer {
        private List<String> bootstrapServers;
        private Integer batchSize = 16;
        private Integer bufferMemory = 32 * 1024 * 1024;
        private String clientId;
        private Class<?> keySerializer = StringSerializer.class;
        private Class<?> valueSerializer = StringSerializer.class;
        private final Map<String, String> properties = new HashMap<>();


        public Map<String, Object> buildProperties() {
            Map<String, Object> properties = new HashMap<>();
            properties.put("bootstrap.servers", this.getBootstrapServers());
            properties.put("client.id", this.getClientId());
            properties.put("batch.size", this.getBatchSize());
            properties.put("buffer.memory", this.getBufferMemory());
            properties.put("key.serializer", this.getKeySerializer());
            properties.put("value.serializer", this.getValueSerializer());
            properties.putAll(this.getProperties());
            return properties;
        }
    }

    @Data
    public static class Listener {
        private Type type = Type.SINGLE;
        private String clientId;
        private Duration pollTimeout;

        public static enum Type {
            SINGLE,
            BATCH;
        }
    }
}
